package net.opentsdb.tsd;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.Atomics;
import com.stumbleupon.async.Deferred;
import net.opentsdb.core.TSDB;
import net.opentsdb.stats.Version;
import net.opentsdb.utils.PluginLoader;
import net.opentsdb.utils.TsdStrUtil;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.handler.codec.http.HttpMethod;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @author Hujf
 * @title: RpcManager
 * @date 2021-09-14 11:00
 * @description: RPC 管理器
 */
public final class RpcManager {
    private static final Logger LOG = LoggerFactory.getLogger(RpcManager.class);

    private static final AtomicReference<RpcManager> INSTANCE = new AtomicReference();

    /**
     * List of activated RPC plugins
     */
    private ImmutableList<RpcPlugin> rpc_plugins;

    private ImmutableMap<String, TelnetRpc> telnet_commands;
    private ImmutableMap<String, HttpRpc> http_commands;

    private TSDB tsdb;

    public RpcManager(TSDB tsdb) {
        this.tsdb = tsdb;
    }

    public static synchronized boolean isInitialized() {
        return INSTANCE.get() != null;
    }

    public static synchronized RpcManager instance(TSDB tsdb) {
        RpcManager exiting = INSTANCE.get();
        if (exiting != null) {
            return exiting;
        }
        RpcManager manager = new RpcManager(tsdb);

        ImmutableList.Builder rpcBulider = ImmutableList.builder();
        if (tsdb.getConfig().hasProperty("tsd.rpc.plugins")) {
            String[] plugins = tsdb.getConfig().getString("tsd.http.rpc.plugins").split(",");
            manager.initializeRpcPlugins(plugins, rpcBulider);
        }

        manager.rpc_plugins = rpcBulider.build();

        ImmutableMap.Builder<String, TelnetRpc> telnetBuilder = ImmutableMap.builder();
        ImmutableMap.Builder<String, HttpRpc> httpBuilder = ImmutableMap.builder();
        manager.initializeBuiltinRpcs(tsdb.getMode(), telnetBuilder, httpBuilder);

        manager.telnet_commands = telnetBuilder.build();
        manager.http_commands = httpBuilder.build();

        INSTANCE.set(manager);
        return manager;
    }

    private void initializeBuiltinRpcs(TSDB.OperationMode mode, ImmutableMap.Builder<String, TelnetRpc> telnet, ImmutableMap.Builder<String, HttpRpc> http) {
        final Boolean enableApi = tsdb.getConfig().getString("tsd.core.enable_api").equals("true");
        final Boolean enableUi = tsdb.getConfig().getString("tsd.core.enable_ui").equals("true");
        final Boolean enableDieDieDie = tsdb.getConfig().getString("tsd.no_diediedie").equals("false");

        LOG.info("Mode: {}, HTTP UI Enabled: {}, HTTP API Enabled: {}", mode, enableUi, enableApi);

        StatsRpc stats = new StatsRpc();
//        ListAggregators aggregators = new ListAggregators();
//        DropCachesRpc dropcaches = new DropCachesRpc();
       final Version version = new Version();

        PutDataPointRpc put = new PutDataPointRpc(tsdb.getConfig());
        telnet.put("exit", new Exit());
        telnet.put("help", new Help());
        telnet.put("stats", stats);
        telnet.put("put", put);

        if (enableApi) {
            http.put("/api/stats", stats);
            http.put("/api/version", version);
            http.put("/api/put?async", put);
            http.put("/api/query",new QueryRpc());
        }

    }

    private void initializeRpcPlugins(String[] plugins, ImmutableList.Builder<RpcPlugin> rpcBuliders) {
        for (String plugin : plugins) {
            RpcPlugin rpc = createAndInitialize(plugin, RpcPlugin.class);
            rpcBuliders.add(rpc);
        }
    }

    private <T> T createAndInitialize(String plugin, Class<T> rpcPluginClass) {
        T instance = PluginLoader.loadSpecificPlugin(plugin, rpcPluginClass);
        Preconditions.checkState(instance != null, "Unable to locate %s using name '%s", plugin, rpcPluginClass);
        try {
            Method initMeth = instance.getClass().getMethod("initialize", TSDB.class);
        } catch (NoSuchMethodException e) {
            e.printStackTrace();
        }
        return instance;
    }

    public Deferred<ArrayList<Object>> shutdown() {

        INSTANCE.set(null);

        final Collection<Deferred<Object>> deferreds = Lists.newArrayList();


        if (rpc_plugins != null) {
            for (final RpcPlugin rpc : rpc_plugins) {
                deferreds.add(rpc.shutdown());
            }
        }

        return Deferred.groupInOrder(deferreds);
    }

    public TelnetRpc lookupTelnetRpc(String command) {
        return telnet_commands.get(command);
    }

    @VisibleForTesting
    protected static final String PLUGIN_BASE_WEBPATH = "plugin";

    private static final Splitter WEBPATH_SPLITTER = Splitter.on('/').trimResults().omitEmptyStrings();


    public boolean isHttpRpcPluginPath(String uri) {
        if (uri.length() <= PLUGIN_BASE_WEBPATH.length()) {
            return false;
        }else{
            int qmark = uri.indexOf("?");
            String path = uri;
            if(qmark!=-1){
                path = uri.substring(0,qmark);
            }

            final List<String> parts = WEBPATH_SPLITTER.splitToList(path);
            return (parts.size() > 1 && parts.get(0).equals(PLUGIN_BASE_WEBPATH));
        }

    }

    public HttpRpc lookupHttpRpc(String route) {
        return http_commands.get(route);
    }

    public class Exit implements TelnetRpc {
        @Override
        public Deferred<Object> execute(TSDB tsdb, Channel chan, String[] command) {
            chan.disconnect();
            return Deferred.fromResult(null);
        }
    }


    public class Help implements TelnetRpc {
        @Override
        public Deferred<Object> execute(TSDB tsdb, Channel chan, String[] command) {
            final StringBuilder buf = new StringBuilder();
            buf.append("There are " + telnet_commands.size() + " available commands: ");
            for (final String commands : telnet_commands.keySet()) {
                buf.append(commands).append(' ');
            }
            buf.append('\n');
            chan.write(buf.toString());
            return Deferred.fromResult(null);
        }
    }
}
